﻿using System.Linq;
using Serilog;
using Nest;
using MongoDB.Driver;
using HYService.RabbitMQ;
using Elasticsearch.Synchronize.Monitor.Entity;
using Elasticsearch.Synchronize.Monitor.Settings;

namespace Elasticsearch.Synchronize.Monitor.Job
{
    public class EventJoinJob : BaseProcessor<SyncJoinByEventIdCfgItem>
    {
        private readonly EventJoinSettings _settings = null;

        public EventJoinJob(EventJoinSettings settings) : base(settings)
        {
            _settings = settings;
        }

        public void Scan()
        {
            var response = _elasticClient.Search<SyncJoinByEventIdCfgItem>(s =>
                s.Query(q => q.Term(t =>
                            t.Field(new Field("State")).Value(MonitorEnum.SyncState.WaitingSync)
                        )
                    ).Index(_settings.ElasticsearchIndex)
                    .Type(_settings.ElasticsearchType).Size(1000)
            );

            if (response.IsValid && response.Documents.Any())
            {
                var docs = response.Documents.ToList();

                RabbitHub.Setup(_settings.SyncJoinByEventIdExchange, ExchangeType.Fanout,
                    _settings.SyncJoinByEventIdQueue);

                RabbitHub.Broadcast<string>(_settings.SyncJoinByEventIdExchange,
                    docs.Select(t => t.EventId.ToString()));
            }
        }

        public void Finish()
        {
            RabbitHub.Subscribe<string>(_settings.WriteSynchronizedTagQueue, (eventId) =>
            {
                Log.Logger.Information(string.Format("write synchronized tag, eventId [{0}]", eventId));
                var _filter = Builders<SyncJoinByEventIdCfgItem>.Filter.Eq("EventId", eventId);
                var _update =
                    Builders<SyncJoinByEventIdCfgItem>.Update.Set("State", MonitorEnum.SyncState.Synchronized);
                var result = _mongoCollection.UpdateOne(_filter, _update);
            });
        }
    }
}